package org.shj.spark.core;

import java.util.Arrays;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

public class CogroupTest {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf();
		conf.setAppName("CogroupTest");
		conf.setMaster("local");
		
		JavaSparkContext sc = new JavaSparkContext(conf);
		
		List<Tuple2<String, String>> names = Arrays.asList(
				new Tuple2<String, String>("s1", "xuruyun"),
				new Tuple2<String, String>("s2", "yangmi"),
				new Tuple2<String, String>("s3", "zhuyin"));
		
		List<Tuple2<String, Integer>> scores = Arrays.asList(
				new Tuple2<String, Integer>("s1", 90),
				new Tuple2<String, Integer>("s2", 80),
				new Tuple2<String, Integer>("s3", 88),
				new Tuple2<String, Integer>("s2", 85),
				new Tuple2<String, Integer>("s3", 87),
				new Tuple2<String, Integer>("s3", 98));
		
		JavaPairRDD<String, String> nameRdd = sc.parallelizePairs(names);
		JavaPairRDD<String, Integer> scorePair = sc.parallelizePairs(scores);
		
		JavaPairRDD<String, Tuple2<Iterable<String>, Iterable<Integer>>> nameScore = nameRdd.cogroup(scorePair);
		
		nameScore.foreach(new VoidFunction<Tuple2<String,Tuple2<Iterable<String>,Iterable<Integer>>>>() {
			private static final long serialVersionUID = -6091531879443619658L;

			public void call(Tuple2<String, Tuple2<Iterable<String>, Iterable<Integer>>> t) throws Exception {
				System.out.println("id=" + t._1 + "; name="+t._2._1 + "; score=" + t._2._2);
			}
		});
		
		sc.close();
	}

}
